"""Stdlib HTTP client for the hubs: device flow + validate + catalog + download. Host-free on purpose — the whole login flow (`login_flow`) is a plain function driven by callbacks, so it runs on a worker thread inside Blender (auth.py) or a C4DThread inside Cinema 4D, and synchronously under a stub HTTP server in tests. Every DCC bundles Python with `ssl`, so `urllib` is enough — no third-party deps. Everything host-shaped (client id, validate `tool`, catalog path) is a parameter; the host adapter binds them once (see tb_core/__init__.py). """ from __future__ import annotations import json import os import shutil import ssl import time import urllib.error import urllib.parse import urllib.request from pathlib import Path _TERMINAL_ERRORS = { "access_denied": "Sign-in was denied in the browser.", "expired_token": "The sign-in request expired — try again.", "invalid_grant": "The sign-in request is no longer valid — try again.", } # Cinema 4D's bundled Python 3.11 ships an OpenSSL compiled for the build # machine's `/usr/local/ssl/cert.pem` and no certifi, so its default trust store # is EMPTY and every HTTPS call dies with CERTIFICATE_VERIFY_FAILED (verified # live in 2026.3). Fall back to the platform's CA bundle when that happens. _CA_BUNDLES = ( "/etc/ssl/cert.pem", # macOS "/etc/pki/tls/certs/ca-bundle.crt", # RHEL/Fedora "/etc/ssl/certs/ca-certificates.crt", # Debian/Ubuntu ) def ca_bundle_path(env=None, exists=os.path.exists): """The first CA bundle worth loading, or None when the host has one already.""" env = os.environ if env is None else env explicit = env.get("SSL_CERT_FILE") if explicit and exists(explicit): return explicit return next((path for path in _CA_BUNDLES if exists(path)), None) def ssl_context(): """A verifying context that works even on a DCC Python with no trust store.""" context = ssl.create_default_context() try: loaded = bool(context.get_ca_certs()) except Exception: loaded = True # can't introspect — trust the interpreter's own store if not loaded: fallback = ca_bundle_path() if fallback: try: context.load_verify_locations(cafile=fallback) except OSError: pass # keep verification ON: a broken bundle must fail loudly return context class _SurfaceRedirects(urllib.request.HTTPRedirectHandler): """Don't auto-follow: urllib turns a redirected POST into a bodyless GET on 301/302 (the apex→www redirect breaks the device flow that way). Surface the 3xx so _send can re-issue the SAME method + body.""" def redirect_request(self, req, fp, code, msg, headers, newurl): return None _OPENER = urllib.request.build_opener( _SurfaceRedirects, urllib.request.HTTPSHandler(context=ssl_context()) ) def _same_site(old_url: str, new_url: str) -> bool: """Is the redirect target still OUR site? `www.host` and `host` are one site: prod 301s www.threejs-blocks.com → the apex, and dropping the bearer on that hop makes every gated download answer 401 "Missing credential" (verified live — DEFAULT_SITE is the www host). A signed S3 url is NOT the same site.""" old, new = urllib.parse.urlsplit(old_url), urllib.parse.urlsplit(new_url) return old.scheme == new.scheme and old.netloc.removeprefix("www.") == new.netloc.removeprefix("www.") class _SafeDownloadRedirects(urllib.request.HTTPRedirectHandler): """Follow the signed artifact redirect without sending the account token to S3 (or any other origin).""" def redirect_request(self, request, fp, code, message, headers, new_url): redirected = super().redirect_request(request, fp, code, message, headers, new_url) if redirected is not None and not _same_site(request.full_url, new_url): redirected.remove_header("Authorization") return redirected _DOWNLOAD_OPENER = urllib.request.build_opener( _SafeDownloadRedirects, urllib.request.HTTPSHandler(context=ssl_context()) ) def _send(request: urllib.request.Request, timeout: float): for _ in range(80): try: with _OPENER.open(request, timeout=timeout) as response: body = response.read().decode("utf-8") return response.status, json.loads(body) if body else {} except urllib.error.HTTPError as error: # 3xx (surfaced) or non-2xx location = error.headers.get("location") if error.code in (301, 302, 307, 308) else None if location: # `request.headers` ONLY — never `header_items()`, which also # returns urllib's unredirected headers (Host, Content-*) # computed for the OLD url. Carrying `Host: www.…` to the apex # makes the server redirect again, forever (verified against # prod's www→apex 301: every hub call died at the cap). request = urllib.request.Request( urllib.parse.urljoin(request.full_url, location), data=request.data, method=request.get_method(), headers=dict(request.headers), ) continue body = error.read().decode("utf-8", "replace") try: return error.code, json.loads(body) if body else {} except ValueError: return error.code, {"raw": body} return 508, {"error": "Too many redirects"} def post_json(site: str, path: str, payload: dict, token: str | None = None, timeout: float = 15.0): """POST JSON, return (status, parsed-body). Network errors raise.""" request = urllib.request.Request(f"{site}{path}", data=json.dumps(payload).encode("utf-8"), method="POST") request.add_header("accept", "application/json") request.add_header("content-type", "application/json") if token: request.add_header("authorization", f"Bearer {token}") return _send(request, timeout) def get_json(site: str, path: str, token: str | None = None, timeout: float = 15.0): """GET JSON, return (status, parsed-body). Network errors raise.""" request = urllib.request.Request(f"{site}{path}", method="GET") request.add_header("accept", "application/json") if token: request.add_header("authorization", f"Bearer {token}") return _send(request, timeout) def validate(site: str, token: str, *, tool: str, tool_version: str | None = None, label: str | None = None): """The same verdict the CLI reads: (status, {entitled, plan, planDisplay, user, credential, …}). `tool` is telemetry-only server-side ("blender-hub" / "c4d-hub").""" return post_json( site, "/api/tools/validate", {"tool": tool, "toolVersion": tool_version, "label": label}, token=token, ) def catalog(site: str, token: str | None = None, *, path: str): """The store feed: (status, {hub: {version}, addons: [{id, entitled, …}]}). Public route, but the per-addon `entitled` flags are only real when the stored credential rides along — anonymous calls get everything locked. """ return get_json(site, path, token=token) def download_artifact( site: str, artifact: str, token: str, destination: str | Path, platform_tag: str = "any", ) -> Path: """Download one gated artifact, following the signed redirect safely.""" path = Path(destination) query = urllib.parse.urlencode({"platform": platform_tag}) name = urllib.parse.quote(artifact, safe="") request = urllib.request.Request( f"{site}/api/tools/download/{name}?{query}", headers={"Accept": "application/octet-stream", "Authorization": f"Bearer {token}"}, ) try: with _DOWNLOAD_OPENER.open(request, timeout=60) as response, path.open("wb") as output: shutil.copyfileobj(response, output) except urllib.error.HTTPError as error: path.unlink(missing_ok=True) body = error.read().decode("utf-8", "replace") try: payload = json.loads(body) detail = payload.get("error") if isinstance(payload, dict) else body except ValueError: detail = body raise RuntimeError(detail or f"Download failed ({error.code}).") from error except Exception: path.unlink(missing_ok=True) raise return path def display_name(verdict: dict | None) -> str | None: """Who is signed in, from a validate verdict: account name > email > credential (device) name. The panel's job is to name the USER — the credential name is a machine label like "Renauds-MBP.local".""" if not isinstance(verdict, dict): return None user = verdict.get("user") or {} credential = verdict.get("credential") or {} return user.get("name") or user.get("email") or credential.get("name") or None def login_flow( site, emit, should_stop=lambda: False, sleep=time.sleep, *, client_id: str, device_name: str, validate_credential=None, ): """Run the whole device flow, reporting through `emit(event_tuple)`. Events: ("open_url", url) → ("token", {token, name, verdict}) on success, or ("error", message) / ("cancelled",). Never raises — the host drains these from a queue on its main thread (Blender modal timer / C4D CoreMessage). `validate_credential(site, token) -> (status, verdict)` is the host's bound `validate`; omit it to win the token without naming the account. """ try: status, flow = post_json( site, "/api/cli/device-code", {"client_id": client_id, "name": device_name}, ) if status != 200 or not flow.get("device_code"): detail = flow.get("error") or f"Could not start sign-in ({status}) at {site}." if status in (404, 405): detail += " Check the site override / TB_SITE_URL." emit(("error", detail)) return emit(("open_url", flow.get("verification_uri_complete") or flow.get("verification_uri"))) interval = float(flow.get("interval") or 2) deadline = time.monotonic() + float(flow.get("expires_in") or 900) while time.monotonic() < deadline: if should_stop(): emit(("cancelled",)) return status, data = post_json(site, "/api/cli/token", {"device_code": flow["device_code"]}) if status == 200 and data.get("access_token"): token = data["access_token"] name = None verdict = None try: # best-effort: the name + plan for the panel; the token is already won if validate_credential is not None: v_status, v_data = validate_credential(site, token) if v_status == 200: verdict = v_data name = display_name(v_data) except Exception: pass emit(("token", {"token": token, "name": name, "verdict": verdict})) return code = data.get("error") if code == "authorization_pending": pass elif code == "slow_down": # defensive — the server never emits it today interval += 1 else: emit(("error", _TERMINAL_ERRORS.get(code, code or f"Sign-in failed ({status})."))) return slept = 0.0 while slept < interval: # sleep in slices so Esc cancels promptly if should_stop(): emit(("cancelled",)) return sleep(0.2) slept += 0.2 emit(("error", _TERMINAL_ERRORS["expired_token"])) except Exception as exc: # URLError, timeout, bad JSON — anything emit(("error", f"Could not reach {site} — {exc}"))